package org.example.rabbitmq.concurrent.mq;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.rabbitmq.client.Channel;
import org.example.rabbitmq.concurrent.lock.RateLimiter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.beans.factory.annotation.Autowired;

import java.io.IOException;

public class ConcurrentTestConsumer implements ChannelAwareMessageListener {
    Logger logger = LoggerFactory.getLogger(ConcurrentTestConsumer.class);
    @Autowired
    ObjectMapper objectMapper;
    @Autowired
    RateLimiter rateLimiter;

    @Override
    public void onMessage(Message message, Channel channel) throws Exception {
        String queueName = message.getMessageProperties().getConsumerQueue();
        try {
            rateLimiter.acquire(queueName);
        } catch (Exception ex) {
            onFailed(ex, queueName, message, channel);
        }
        try {
            String prop = objectMapper.writeValueAsString(message.getMessageProperties());
            logger.info("channel-" + channel.getChannelNumber() + " " + queueName + " " + new String(message.getBody()) + " consumerTag: " + message.getMessageProperties().getConsumerTag() + " props: " + prop);
            Thread.sleep(10000);
            channel.basicAck(message.getMessageProperties().getDeliveryTag(), false);
            rateLimiter.release(queueName);
        } catch (Exception ex) {
            onFailed(ex, queueName, message, channel);
        }
    }

    private void onFailed(Exception ex, String queueName, Message message, Channel channel) {
        rateLimiter.release(queueName);
        try {
            channel.basicReject(message.getMessageProperties().getDeliveryTag(), true);
        } catch (IOException e) {
            logger.error("reject failed, queueName: " + queueName + " consumeTag: " + message.getMessageProperties().getConsumerTag(), e);
            // channel.basicRecover(false);
        }
    }
}